/********************************************
 * 功能说明: 
 * 模块名称: 
 * 系统名称: 
 * 软件版权: Frank
 * 系统版本: 1.0.0
 * 开发人员: Frank
 * 开发时间: 2020/3/5 20:28
 * 审核人员: 
 * 相关文档: 
 * 修改记录: 修改日期 修改人员 修改说明
 *********************************************/
package com.spring.arch.common.jms.publish;

import com.hazelcast.core.Hazelcast;
import com.hazelcast.core.HazelcastInstance;
import com.hazelcast.core.ITopic;
import com.spring.arch.common.utils.ExecutorUtil;
import com.spring.arch.common.utils.NamedThreadFactory;
import com.spring.arch.common.jms.Message;
import com.spring.arch.common.utils.ExecutorUtil;
import com.spring.arch.common.utils.NamedThreadFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;

/**
 * 消息发布者
 *
 * @author Frank
 * @version 1.0.0.1
 * @since 2020/3/5 20:28
 */
public class MessagePublisher implements Runnable {

    private final Logger LOGGER = LoggerFactory.getLogger(this.getClass());

    /**
     * Topic线程池，为每一个topic分配一个新线程
     */
    private static ThreadPoolExecutor executor = new ThreadPoolExecutor(20,Integer.MAX_VALUE,
            60L,TimeUnit.SECONDS,
            new LinkedBlockingQueue<>(1000), new NamedThreadFactory("Hazelcast-jms"));

    private final AtomicBoolean destroyed;

    private volatile boolean running;

    private ArrayBlockingQueue<Message> queues = new ArrayBlockingQueue<>(20);

    private final ITopic<Message> topic;

    private final HazelcastInstance hzInstance;

    public MessagePublisher(String topicName) {
        this.destroyed = new AtomicBoolean(false);
        this.hzInstance = Hazelcast.newHazelcastInstance();
        this.topic = createTopic(topicName);
        this.running = true;
        executor.submit(this);
        registerShutdownHook();
    }

    public MessagePublisher(HazelcastInstance hzInstance, String topicName) {
        this.destroyed = new AtomicBoolean(false);
        this.hzInstance = hzInstance;
        this.topic = createTopic(topicName);
        this.running = true;
        executor.submit(this);
        registerShutdownHook();
    }

    @Override
    public void run() {
        do {
            send();
            sleep();
        } while (running);
    }

    public void publish(Message message) {
        if (message == null) {
            LOGGER.warn("发布消息为空");
            return;
        }
        // 阻塞写
        try {
            boolean res = queues.offer(message, 100, TimeUnit.MILLISECONDS);
            if (!res) {
                LOGGER.error("[线程{}-topic{}-message{}]发布失败，当前队列大小:{}", Thread.currentThread().getName(), topic.getName(), message, queues.size());
            }
        } catch (InterruptedException e) {
            LOGGER.error("[线程{}-topic{}]-中断异常", Thread.currentThread().getName(), topic.getName(), e);
            Thread.currentThread().interrupt();
        }
        LOGGER.info("[线程{}-topic{}-message{}]发布成功", Thread.currentThread().getName(), topic.getName(), message);
    }

    public void stop() {
        running = false;
    }

    private void send() {
        // 非阻塞读
        Message message = queues.poll();
        if (message != null) {
            topic.publish(message);
        }
        LOGGER.info("[线程{}-topic{}]队列中元素数量={}", Thread.currentThread().getName(),topic.getName(), queues.size());
    }

    private void sleep() {
        try {
            TimeUnit.SECONDS.sleep(2);
        } catch (InterruptedException e) {
            LOGGER.error("[线程{}-topic{}]-中断异常", Thread.currentThread().getName(), topic.getName(), e);
            Thread.currentThread().interrupt();
        }
    }

    private ITopic<Message> createTopic(String topicName) {
        return hzInstance.getTopic(topicName);
    }

    private void registerShutdownHook() {
        Runtime.getRuntime().addShutdownHook(new Thread(() -> {
            if (!destroyed.compareAndSet(false, true)) {
                return;
            }
            // 线程暂停
            running = false;
            // 清空消息队列
            queues.clear();
            // 关闭线程池
            ExecutorUtil.gracefulShutdown(executor, 1000);
        }));
    }
}
